/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.openness.crawler.crawler;

import java.util.ArrayList;
import java.util.List;

import org.apache.http.HttpStatus;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.openness.crawler.fetcher.CustomFetchStatus;
import com.openness.crawler.fetcher.PageFetchResult;
import com.openness.crawler.fetcher.PageFetcher;
import com.openness.crawler.frontier.DocIDServer;
import com.openness.crawler.frontier.Frontier;
import com.openness.crawler.parser.HtmlParseData;
import com.openness.crawler.parser.ParseData;
import com.openness.crawler.parser.Parser;
import com.openness.crawler.robotstxt.RobotstxtServer;
import com.openness.crawler.url.WebURL;
import com.openness.crawler.util.Constants;
import com.openness.crawler.util.Util;
import com.openness.crawler.zookeeper.ProtocolSupport;
import com.openness.crawler.zookeeper.ZooKeeperOperation;
import com.openness.example.crawler.Statistic;

/**
 * WebCrawler class in the Runnable class that is executed by each crawler
 * thread.
 * 
 * @author Yasser Ganjisaffar <lastname at gmail dot com>
 */
public class WebCrawler extends ProtocolSupport implements Runnable {

	protected static final Logger LOGGER = LoggerFactory
			.getLogger(WebCrawler.class);

	/**
	 * The id associated to the crawler thread running this instance
	 */
	protected int myId;

	/**
	 * The controller instance that has created this crawler thread. This
	 * reference to the controller can be used for getting configurations of the
	 * current crawl or adding new seeds during runtime.
	 */
	protected CrawlController myController;

	/**
	 * The thread within which this crawler instance is running.
	 */
	private Thread myThread;

	/**
	 * The parser that is used by this crawler instance to parse the content of
	 * the fetched pages.
	 */
	private Parser parser;

	/**
	 * The fetcher that is used by this crawler instance to fetch the content of
	 * pages from the web.
	 */
	private PageFetcher pageFetcher;

	/**
	 * The RobotstxtServer instance that is used by this crawler instance to
	 * determine whether the crawler is allowed to crawl the content of each
	 * page.
	 */
	private RobotstxtServer robotstxtServer;

	/**
	 * The DocIDServer that is used by this crawler instance to map each URL to
	 * a unique docid.
	 */
	private DocIDServer docIdServer;

	/**
	 * The Frontier object that manages the crawl queue.
	 */
	private Frontier frontier;

	private ZooKeeper zk;

	private String crawlingThread;

	private boolean requestFinish = false;

	private boolean finish = false;

	/**
	 * Initializes the current instance of the crawler
	 * 
	 * @param id
	 *            the id of this crawler instance
	 * @param crawlController
	 *            the controller that manages this crawling session
	 */
	public void init(int id, CrawlController crawlController, ZooKeeper zk) {
		this.myId = id;

		this.pageFetcher = crawlController.getPageFetcher();

		this.robotstxtServer = crawlController.getRobotstxtServer();

		this.docIdServer = crawlController.getDocIdServer();

		this.frontier = crawlController.getFrontier();

		this.parser = new Parser(crawlController.getConfig());

		this.myController = crawlController;

		this.zk = zk;
	}

	/**
	 * Get the id of the current crawler instance
	 * 
	 * @return the id of the current crawler instance
	 */
	public int getMyId() {
		return myId;
	}

	public CrawlController getMyController() {
		return myController;
	}

	/**
	 * This function is called just before starting the crawl by this crawler
	 * instance. It can be used for setting up the data structures or
	 * initializations needed by this crawler instance.
	 */
	public void onStart() {
		// Do nothing by default
		// Sub-classed can override this to add their custom functionality
	}

	/**
	 * This function is called just before the termination of the current
	 * crawler instance. It can be used for persisting in-memory data or other
	 * finalization tasks.
	 */
	public void onBeforeExit() {
		// Do nothing by default
		// Sub-classed can override this to add their custom functionality
	}

	/**
	 * This function is called once the header of a page is fetched. It can be
	 * overwritten by sub-classes to perform custom logic for different status
	 * codes. For example, 404 pages can be logged, etc.
	 * 
	 * @param webUrl
	 * @param statusCode
	 * @param statusDescription
	 */
	protected void handlePageStatusCode(WebURL webUrl, int statusCode,
			String statusDescription) {
		// Do nothing by default
		// Sub-classed can override this to add their custom functionality
	}

	/**
	 * This function is called if the content of a url could not be fetched.
	 * 
	 * @param webUrl
	 */
	protected void onContentFetchError(WebURL webUrl) {
		// Do nothing by default
		// Sub-classed can override this to add their custom functionality
	}

	/**
	 * This function is called if there has been an error in parsing the
	 * content.
	 * 
	 * @param webUrl
	 */
	protected void onParseError(WebURL webUrl) {
		// Do nothing by default
		// Sub-classed can override this to add their custom functionality
	}

	/**
	 * The CrawlController instance that has created this crawler instance will
	 * call this function just before terminating this crawler thread. Classes
	 * that extend WebCrawler can override this function to pass their local
	 * data to their controller. The controller then puts these local data in a
	 * List that can then be used for processing the local data of crawlers (if
	 * needed).
	 */
	public Object getMyLocalData() {
		return null;
	}

	public void run() {
		onStart();

		while (true) {
			crawlingThread = null;
			
			List<WebURL> assignedURLs = new ArrayList<WebURL>();

			frontier.getNextURLs(myController.getConfig().getNextUrlSize(),
					assignedURLs);

			if (assignedURLs.size() > 0) {
				LOGGER.debug(Thread.currentThread().getName()
						+ " get next urls size is " + assignedURLs.size());

				try {
					crawlingThread = retryOperation(new ZooKeeperOperation<String>() {

						private String crawlingThread = null;

						@Override
						public String execute() throws Exception {
							if (crawlingThread != null) {
								return crawlingThread;
							}

							List<String> child = zk.getChildren(
									Constants.ZK_CRAWLING_THREAD, false);

							boolean exist = false;

							for (String chirldren : child) {
								byte[] data = zk.getData(
										Constants.ZK_CRAWLING_THREAD
												+ Constants.ZK_SEPARATOR
												+ chirldren, false, null);

								if (data.equals(Util.getData())) {
									crawlingThread = Constants.ZK_CRAWLING_THREAD
											+ Constants.ZK_SEPARATOR
											+ chirldren;

									exist = true;

									break;
								}
							}

							if (!exist) {
								crawlingThread = zk
										.create(Constants.ZK_CRAWLING_THREAD
												+ Constants.ZK_SEPARATOR
												+ Constants.ZK_CRAWLING_THREAD_PREFIX,
												Constants.EMPTY_BYTEARRAY,
												Ids.OPEN_ACL_UNSAFE,
												CreateMode.EPHEMERAL_SEQUENTIAL);
							}

							return crawlingThread;
						}

					});

					List<WebURL> processed = new ArrayList<WebURL>();

					for (WebURL curURL : assignedURLs) {
						if (curURL != null) {
							processPage(curURL);

							processed.add(curURL);
						}

						frontier.setProcessed(processed);

						if (myController.isShuttingDown()) {
							LOGGER.info("Exiting because of controller shutdown.");

							return;
						}
					}
				} catch (Exception e) {
					LOGGER.error("create crawlingthread node error: "
							+ e.toString());
				} finally {
					if (crawlingThread != null) {
						try {
							retryOperation(new ZooKeeperOperation<Object>() {

								@Override
								public Object execute() throws Exception {
									if (zk.exists(crawlingThread, false) != null) {
										zk.delete(crawlingThread, -1);
									}

									return null;
								}

							});
						} catch (Exception e) {
							LOGGER.error("delete crawlingthread node error: "
									+ e.toString());
						}
					}
				}

				Statistic.stat();
			} else {
				LOGGER.debug(Thread.currentThread().getName()
						+ " get next urls size is 0, wait");

				if (requestFinish) {
					finish = true;

					return;
				}

				try {
					Thread.sleep(10000);
				} catch (InterruptedException e) {
				}
			}
		}

	}

	/**
	 * Classes that extends WebCrawler can overwrite this function to tell the
	 * crawler whether the given url should be crawled or not. The following
	 * implementation indicates that all urls should be included in the crawl.
	 * 
	 * @param url
	 *            the url which we are interested to know whether it should be
	 *            included in the crawl or not.
	 * @return if the url should be included in the crawl it returns true,
	 *         otherwise false is returned.
	 */
	public boolean shouldVisit(WebURL url) {
		return true;
	}

	/**
	 * Classes that extends WebCrawler can overwrite this function to process
	 * the content of the fetched and parsed page.
	 * 
	 * @param page
	 *            the page object that is just fetched and parsed.
	 */
	public void visit(Page page) {
		// Do nothing by default
		// Sub-classed can override this to add their custom functionality
	}

	private void processPage(WebURL curURL) {
		if (curURL == null) {
			return;
		}

		PageFetchResult fetchResult = null;

		try {
			fetchResult = pageFetcher.fetchHeader(curURL);

			int statusCode = fetchResult.getStatusCode();

			handlePageStatusCode(curURL, statusCode,
					CustomFetchStatus.getStatusDescription(statusCode));

			/*
			 * statusCode is not ok
			 * 
			 * ******************************************************************
			 */
			if (statusCode != HttpStatus.SC_OK) {
				if (statusCode == HttpStatus.SC_MOVED_PERMANENTLY
						|| statusCode == HttpStatus.SC_MOVED_TEMPORARILY) {
					if (myController.getConfig().isFollowRedirects()) {
						String movedToUrl = fetchResult.getMovedToUrl();

						if (movedToUrl == null) {
							return;
						}

						long newDocId = docIdServer.getDocId(movedToUrl);

						if (newDocId > 0) {
							// Redirect page is already seen
							return;
						}

						WebURL webURL = new WebURL();

						webURL.setURL(movedToUrl);
						webURL.setParentDocid(curURL.getParentDocid());
						webURL.setParentUrl(curURL.getParentUrl());
						webURL.setDepth(curURL.getDepth());
						webURL.setDocid(-1);
						webURL.setAnchor(curURL.getAnchor());

						if (shouldVisit(webURL)
								&& robotstxtServer.allows(webURL)) {
							webURL.setDocid(docIdServer.getNewDocID(movedToUrl));

							frontier.schedule(webURL);
						}
					}
				} else if (fetchResult.getStatusCode() == CustomFetchStatus.PageTooBig) {
					LOGGER.info("Skipping a page which was bigger than max allowed size: "
							+ curURL.getURL());
				}

				return;
			}

			/*
			 * statusCode is ok
			 * 
			 * ******************************************************************
			 */
			if (!curURL.getURL().equals(fetchResult.getFetchedUrl())) {
				if (docIdServer.isSeenBefore(fetchResult.getFetchedUrl())) {
					// Redirect page is already seen
					return;
				}

				curURL.setURL(fetchResult.getFetchedUrl());
				curURL.setDocid(docIdServer.getNewDocID(fetchResult
						.getFetchedUrl()));
			}

			Page page = new Page(curURL);

			long docid = curURL.getDocid();

			// 当前Page的深度值
			int depth = curURL.getDepth();

			// 当前Page中的URL所代表Page的深度值
			int nextDepth = depth + 1;

			// 所要爬取Page的最大深度值
			int maxCrawlDepth = myController.getConfig()
					.getMaxDepthOfCrawling();

			if (!fetchResult.fetchContent(page)) {
				onContentFetchError(curURL);

				return;
			}

			if (!parser.parse(page, curURL.getURL())) {
				onParseError(curURL);

				return;
			}

			ParseData parseData = page.getParseData();

			/*
			 * 处理Html Page内的URL
			 * 
			 * 如果当前Page中的URL所代表的Page的深度值等于或大于MaxDepth，则表示这些URL可以丢弃
			 * 
			 * ******************************************************************
			 */
			if ((maxCrawlDepth == -1 || nextDepth <= maxCrawlDepth)
					&& parseData instanceof HtmlParseData) {
				HtmlParseData htmlParseData = (HtmlParseData) parseData;

				List<WebURL> outgoingUrls = htmlParseData.getOutgoingUrls();

				if (outgoingUrls != null) {
					LOGGER.debug("outgoingUrls size: " + outgoingUrls.size());

					if (outgoingUrls.size() > 0) {
						for (int index = outgoingUrls.size() - 1; index >= 0; index--) {
							if (!outgoingUrls.get(index).getDomain()
									.equals(curURL.getDomain())) {
								outgoingUrls.remove(index);
							}
						}
					}

					LOGGER.debug("same domain size: " + outgoingUrls.size());

					if (outgoingUrls.size() > 0) {
						List<Long> docids = docIdServer.getDocIds(outgoingUrls);

						for (int index = outgoingUrls.size() - 1; index >= 0; index--) {
							if (docids.get(index) > 0
									|| !shouldVisit(outgoingUrls.get(index))
									|| !robotstxtServer.allows(outgoingUrls
											.get(index))) {
								outgoingUrls.remove(index);
							}
						}
					}

					LOGGER.debug("not exist and should visit size: "
							+ outgoingUrls.size());

					// 此时，剩余的URL为需要爬取的URL
					if (outgoingUrls.size() > 0) {
						for (WebURL url : outgoingUrls) {
							url.setParentDocid(docid);

							url.setParentUrl(curURL.getURL());

							url.setDepth((short) nextDepth);
						}

						// 为这些URL分配docid
						List<Long> newDocIds = docIdServer
								.getNewDocIDs(outgoingUrls);

						for (int index = 0; index < outgoingUrls.size(); index++) {
							outgoingUrls.get(index).setDocid(
									newDocIds.get(index));
						}

						for (int index = outgoingUrls.size() - 1; index >= 0; index--) {
							if (outgoingUrls.get(index).getDocid() <= 0) {
								outgoingUrls.remove(index);
							}
						}
					}

					LOGGER.debug("asign new docid size: " + outgoingUrls.size());

					frontier.scheduleAll(outgoingUrls);

					LOGGER.debug("schedule size: " + outgoingUrls.size());
				}
			}

			try {
				visit(page);
			} catch (Exception e) {
				LOGGER.error("Exception while running the visit method. Message: '"
						+ e.getMessage() + "' at " + e.getStackTrace()[0]);
			}
		} catch (Exception e) {
			LOGGER.error(e.getMessage() + ", while processing: "
					+ curURL.getURL());
		} finally {
			if (fetchResult != null) {
				fetchResult.discardContentIfNotConsumed();
			}
		}
	}

	public Thread getThread() {
		return myThread;
	}

	public void setThread(Thread myThread) {
		this.myThread = myThread;
	}

	public void requestFinish() {
		this.requestFinish = true;
	}

	public boolean isFinished() {
		return finish;
	}

}
